package com.xiaochong.bitong.job.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xc.logclient.utils.LogTrace;
import com.xiaochong.bitong.job.config.hbase.CurrencyInfoHbaseComponent;
import com.xiaochong.bitong.job.config.hbase.HbaseBaseQueryComponent;
import com.xiaochong.service.common.config.redis.RedisServer;
import com.xiaochong.tir.common.data.hbasepo.TradesHistoryCountFive;
import com.xiaochong.tir.common.data.hbasepo.TradesToKLine;
import com.xiaochong.tir.common.data.mapper.ExchangeInfoMapper;
import com.xiaochong.tir.common.data.mapper.ExchangeTransactionpairMiddleMapper;
import com.xiaochong.tir.common.data.mapper.FundsFlowMapper;
import com.xiaochong.tir.common.data.mapperSearchResult.ExchangeTransactionpair;
import com.xiaochong.tir.common.data.po.ExchangeInfo;
import com.xiaochong.tir.common.data.po.ExchangeTransactionpairMiddle;
import com.xiaochong.tir.common.data.po.FundsFlow;
import com.xiaochong.tir.common.data.po.KLine;
import com.xiaochong.tir.common.util.DateUtils;
import com.xiaochong.tir.common.util.HbaseTableUtils;
import com.xiaochong.tir.common.util.RedisKeyApp;
import com.xiaochong.tir.common.util.RedisKeysUtils;
import net.sf.json.util.JSONTokener;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.*;

/**
 * @Description: java类作用描述
 * @Author: org
 * @CreateDate: 2018/8/17 11:02
 */
@Service
public class KActionSectionService {

    @Autowired
    private ExchangeInfoMapper exchangeInfoMapper;
    @Autowired
    private ExchangeTransactionpairMiddleMapper exchangeTransactionpairMiddleMapper;
    @Autowired
    private HbaseBaseQueryComponent hbaseBaseQueryComponent;
    @Autowired
    private CurrencyInfoHbaseComponent currencyInfoHbaseComponent;
    @Autowired
    private FundsFlowMapper fundsFlowMapper;
    @Autowired
    private RedisServer redisServer;

    /**
     *计算5分钟k线数据
     */
    public void calculateFiveMKAction() {
        try {
            //获取交易所-》交易对-》判断5/15/30/1h/4h-》分别计算维度k线
            ExchangeInfo exchangeInfo=new ExchangeInfo();
            exchangeInfo.setEnableFlag( true );
            exchangeInfo.setDeleteFlag( false );
            exchangeInfo.setKlineStatus( true );
            LocalDateTime time = LocalDateTime.now();
            List<ExchangeInfo> infoList=exchangeInfoMapper.selectByEntityList( exchangeInfo );
            if(infoList != null && infoList.size() >0){
                for(ExchangeInfo exchange : infoList){
                    ExchangeTransactionpairMiddle query = new ExchangeTransactionpairMiddle();
                    query.setExchangeId(exchange.getId());
                    //获取所有货币,组装每个需要诊断数据放到redis中
                    List<ExchangeTransactionpair> middleList = exchangeTransactionpairMiddleMapper.selectByEntityListOfKLine(query);
                    if(middleList != null && middleList.size() > 0){
                        for (ExchangeTransactionpair pair :middleList){
                            calculateFiveAllKAction(pair.getExchangeId(),pair.getTransactionName(),time,exchange.getExchangeName());
                        }
                    }
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            LogTrace.error("计算维度内k线error",e);
        }

    }

    /**
     * 4h的k线计算
     */
    public void calculateFourHKAction() {
        try {
            //获取交易所-》交易对-》判断5/15/30/1h/4h-》分别计算维度k线
            ExchangeInfo exchangeInfo=new ExchangeInfo();
            exchangeInfo.setEnableFlag( true );
            exchangeInfo.setDeleteFlag( false );
            exchangeInfo.setKlineStatus( true );
            LocalDateTime time = LocalDateTime.now();
            List<ExchangeInfo> infoList=exchangeInfoMapper.selectByEntityList( exchangeInfo );
            if(infoList != null && infoList.size() >0){
                for(ExchangeInfo exchange : infoList){
                    ExchangeTransactionpairMiddle query = new ExchangeTransactionpairMiddle();
                    query.setExchangeId(exchange.getId());
                    //获取所有货币,组装每个需要诊断数据放到redis中
                    List<ExchangeTransactionpair> middleList = exchangeTransactionpairMiddleMapper.selectByEntityListOfKLine(query);
                    if(middleList != null && middleList.size() > 0){
                        for (ExchangeTransactionpair pair :middleList){
                            calculateFourHMakeKAction(pair.getExchangeId(),pair.getTransactionName(),time);
                        }
                    }
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    /**
     * 4h的k线图
     * @param exchangeId
     * @param transactionName
     * @param time
     * @throws Exception
     */
    private void calculateFourHMakeKActionNew(Integer exchangeId, String transactionName, LocalDateTime time) throws Exception{
        LocalDateTime oldStart = time.minusHours(4);
        TradesToKLine tradesToKLineNew = new TradesToKLine();
        tradesToKLineNew.setVolume("0");
        tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(oldStart).getTime()));
        // oldStart = time.minusMinutes(60);
        for(int i=0;i<4;i++){
            String rowKey = exchangeId+"@"+transactionName+"@"+DateUtils.localDateTimeFormatString(oldStart, DateUtils.format_ymdh);
            LogTrace.infoKvs("4hK线数据", "rowKey", rowKey);
            String openPriceStr;
            String highOpenStr;
            String lowCloseStr;
            String closePriceStr;
            String volumeStr;
            Map<String, Object> hbaseData = hbaseBaseQueryComponent.selectData(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME
                    , rowKey);
            if(null != hbaseData && hbaseData.get("volume") != null && hbaseData.get("highOpen") != null && hbaseData.get("lowClose") != null) {
                openPriceStr = hbaseData.get("openPrice").toString();
                highOpenStr = hbaseData.get("highOpen").toString();
                lowCloseStr = hbaseData.get("lowClose").toString();
                closePriceStr = hbaseData.get("closePrice").toString();
                volumeStr = hbaseData.get("volume").toString();
                //交易量
                BigDecimal volume = new BigDecimal(hbaseData.get("volume").toString());
                // BigDecimal volume = new BigDecimal(tradesToKLineNew.getVolume());
                volume = volume.add(new BigDecimal(volumeStr));
                tradesToKLineNew.setVolume(volume.toString());
                //开盘
                if (StringUtils.isBlank(tradesToKLineNew.getOpenPrice())) {
                    tradesToKLineNew.setOpenPrice(openPriceStr);
                }
                //收盘
                tradesToKLineNew.setClosePrice(closePriceStr);
                //最高价格
                if (StringUtils.isBlank(tradesToKLineNew.getHighestPrice())) {
                    tradesToKLineNew.setHighestPrice(highOpenStr);
                } else {
                    BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                    BigDecimal highOpen = new BigDecimal(tradesToKLineNew.getHighestPrice());
                    if (bigDecimal.compareTo(highOpen) > 0) {
                        tradesToKLineNew.setHighestPrice(highOpenStr);
                    }
                }
                //最低价格
                if (StringUtils.isBlank(tradesToKLineNew.getLowestPrice())) {
                    tradesToKLineNew.setLowestPrice(lowCloseStr);
                } else {
                    BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                    BigDecimal lowClose = new BigDecimal(tradesToKLineNew.getLowestPrice());
                    if (bigDecimal.compareTo(lowClose) < 0) {
                        tradesToKLineNew.setLowestPrice(lowCloseStr);
                    }
                }
            }
            oldStart = oldStart.plusHours(1);
        }
        if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
            String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_4H+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time.minusHours(4), DateUtils.format_ymdh);
            currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
        }
    }
    /**
     * 4h的k线图
     * @param exchangeId
     * @param transactionName
     * @param time
     * @throws Exception
     */
    private void calculateFourHMakeKAction(Integer exchangeId, String transactionName, LocalDateTime time) throws Exception{
        LocalDateTime oldStart = time.minusHours(4);
        TradesToKLine tradesToKLineNew = new TradesToKLine();
        tradesToKLineNew.setVolume("0");
        tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(oldStart).getTime()));
       // oldStart = time.minusMinutes(60);
        for(int i=0;i<4;i++){
            String rowKey = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1H+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(oldStart, DateUtils.format_ymdh);
            LogTrace.infoKvs("4hK线数据", "rowKey", rowKey);
            String openPriceStr;
            String highOpenStr;
            String lowCloseStr;
            String closePriceStr;
            String volumeStr;
            TradesToKLine tradesToKLine = getHbaseDataByRowKey(rowKey);
            if(tradesToKLine != null){
                openPriceStr = tradesToKLine.getOpenPrice();
                highOpenStr = tradesToKLine.getHighestPrice();
                lowCloseStr = tradesToKLine.getLowestPrice();
                closePriceStr = tradesToKLine.getClosePrice();
                volumeStr = tradesToKLine.getVolume();
                //交易量
                BigDecimal volume = new BigDecimal(tradesToKLineNew.getVolume());
                volume = volume.add(new BigDecimal(volumeStr));
                tradesToKLineNew.setVolume(volume.toString());
                //开盘
                if (StringUtils.isBlank(tradesToKLineNew.getOpenPrice())) {
                    tradesToKLineNew.setOpenPrice(openPriceStr);
                }
                //收盘
                tradesToKLineNew.setClosePrice(closePriceStr);
                //最高价格
                if (StringUtils.isBlank(tradesToKLineNew.getHighestPrice())) {
                    tradesToKLineNew.setHighestPrice(highOpenStr);
                } else {
                    BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                    BigDecimal highOpen = new BigDecimal(tradesToKLineNew.getHighestPrice());
                    if (bigDecimal.compareTo(highOpen) > 0) {
                        tradesToKLineNew.setHighestPrice(highOpenStr);
                    }
                }
                //最低价格
                if (StringUtils.isBlank(tradesToKLineNew.getLowestPrice())) {
                    tradesToKLineNew.setLowestPrice(lowCloseStr);
                } else {
                    BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                    BigDecimal lowClose = new BigDecimal(tradesToKLineNew.getLowestPrice());
                    if (bigDecimal.compareTo(lowClose) < 0) {
                        tradesToKLineNew.setLowestPrice(lowCloseStr);
                    }
                }
            }
            oldStart = oldStart.plusHours(1);
        }
        if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
            String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_4H+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time.minusHours(4), DateUtils.format_ymdh);
            currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
        }
    }

    private void calculateFiveAllKAction(Integer exchangeId, String transactionName,LocalDateTime time,String exchangeName) throws Exception {
        int m = time.getMinute();
        LocalDateTime oldStart = time.minusMinutes(5);
        TradesToKLine tradesToKLineNew = new TradesToKLine();
        tradesToKLineNew.setVolume("0");
        tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(oldStart).getTime()));
        if(exchangeId.intValue() == 16 && "BTC/USDT".equals(transactionName)){
            LogTrace.infoKvs("组装5分钟k线数据", "transactionName", transactionName);
        }
        //计算5分钟上一次的数据
        for(int i=0;i<5;i++){
            String rowKey = exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(oldStart, DateUtils.format_ymdhm);
            String data = hbaseBaseQueryComponent.selectByQualifierData(HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME
                    , HbaseTableUtils.API_SERVICE_K_ACTION_TABLE_NAME_FAMILY
                    , HbaseTableUtils.API_SERVICE_K_ACTION_QUALIFIER
                    , rowKey);
            LogTrace.infoKvs("组装5分钟k线数据", "rowKey", rowKey, "data", data);
            if(StringUtils.isNotBlank(data)) {
                //下标： 0 时间戳 1 开盘 2最高价 3最低  4收盘 5交易量 6涨幅
                LogTrace.infoKvs("一分钟K线数据", "rowKey", rowKey, "data", data);
                String openPriceStr;
                String highOpenStr;
                String lowCloseStr;
                String closePriceStr;
                String volumeStr;
                Object listArray = new JSONTokener(data).nextValue();
                if (listArray instanceof net.sf.json.JSONArray) {
                    JSONArray jsonArray = JSON.parseArray(data);
                    openPriceStr = jsonArray.getString(1);
                    highOpenStr = jsonArray.getString(2);
                    lowCloseStr = jsonArray.getString(3);
                    closePriceStr = jsonArray.getString(4);
                    volumeStr = jsonArray.getString(5);

                } else {
                    JSONObject jsonObject3 = JSON.parseObject(data);
                    TradesToKLine tradesToKLine = jsonObject3.toJavaObject(TradesToKLine.class);
                    openPriceStr = tradesToKLine.getOpenPrice();
                    highOpenStr = tradesToKLine.getHighestPrice();
                    lowCloseStr = tradesToKLine.getLowestPrice();
                    closePriceStr = tradesToKLine.getClosePrice();
                    volumeStr = tradesToKLine.getVolume();
                }

                //交易量
                BigDecimal volume = new BigDecimal(tradesToKLineNew.getVolume());
                volume = volume.add(new BigDecimal(volumeStr));
                tradesToKLineNew.setVolume(volume.toString());

                //开盘
                if (StringUtils.isBlank(tradesToKLineNew.getOpenPrice())) {
                    tradesToKLineNew.setOpenPrice(openPriceStr);
                }
                //收盘
                tradesToKLineNew.setClosePrice(closePriceStr);
                //最高价格
                if (StringUtils.isBlank(tradesToKLineNew.getHighestPrice())) {
                    tradesToKLineNew.setHighestPrice(highOpenStr);
                } else {
                    BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                    BigDecimal highOpen = new BigDecimal(tradesToKLineNew.getHighestPrice());
                    if (bigDecimal.compareTo(highOpen) > 0) {
                        tradesToKLineNew.setHighestPrice(highOpenStr);
                    }
                }
                //最低价格
                if (StringUtils.isBlank(tradesToKLineNew.getLowestPrice())) {
                    tradesToKLineNew.setLowestPrice(lowCloseStr);
                } else {
                    BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                    BigDecimal lowClose = new BigDecimal(tradesToKLineNew.getLowestPrice());
                    if (bigDecimal.compareTo(lowClose) < 0) {
                        tradesToKLineNew.setLowestPrice(lowCloseStr);
                    }
                }
            }
            oldStart =oldStart.plusMinutes(1);
        }
        //保存hbase
        if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
            String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_5M+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time.minusMinutes(5), DateUtils.format_ymdhm);
            currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
            //分别更新15M,30M,1H
            calculateParkMakeKAction(exchangeId,transactionName,time,tradesToKLineNew,exchangeName);
        }
        //如果是如下分钟需要计算
        if(m%15 == 0){
            tradesToKLineNew = new TradesToKLine();
            oldStart = time.minusMinutes(15);
            tradesToKLineNew.setVolume("0");
            tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(oldStart).getTime()));
            for(int i=0;i<3;i++){
                String rowKey = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_5M+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(oldStart, DateUtils.format_ymdhm);
                LogTrace.infoKvs("5分钟K线数据", "rowKey", rowKey);
                String openPriceStr;
                String highOpenStr;
                String lowCloseStr;
                String closePriceStr;
                String volumeStr;
                /*if(exchangeId.intValue() == 11 && "ETH/BTC".equals(transactionName)){
                    LogTrace.infoKvs("5分钟K线数据", "transactionName", transactionName);
                }*/
                TradesToKLine tradesToKLine = getHbaseDataByRowKey(rowKey);
                if(tradesToKLine != null){
                    openPriceStr = tradesToKLine.getOpenPrice();
                    highOpenStr = tradesToKLine.getHighestPrice();
                    lowCloseStr = tradesToKLine.getLowestPrice();
                    closePriceStr = tradesToKLine.getClosePrice();
                    volumeStr = tradesToKLine.getVolume();
                    //交易量
                    BigDecimal volume = new BigDecimal(tradesToKLineNew.getVolume());
                    volume = volume.add(new BigDecimal(volumeStr));
                    tradesToKLineNew.setVolume(volume.toString());
                    //开盘
                    if (StringUtils.isBlank(tradesToKLineNew.getOpenPrice())) {
                        tradesToKLineNew.setOpenPrice(openPriceStr);
                    }
                    //收盘
                    tradesToKLineNew.setClosePrice(closePriceStr);
                    //最高价格
                    if (StringUtils.isBlank(tradesToKLineNew.getHighestPrice())) {
                        tradesToKLineNew.setHighestPrice(highOpenStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                        BigDecimal highOpen = new BigDecimal(tradesToKLineNew.getHighestPrice());
                        if (bigDecimal.compareTo(highOpen) > 0) {
                            tradesToKLineNew.setHighestPrice(highOpenStr);
                        }
                    }
                    //最低价格
                    if (StringUtils.isBlank(tradesToKLineNew.getLowestPrice())) {
                        tradesToKLineNew.setLowestPrice(lowCloseStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                        BigDecimal lowClose = new BigDecimal(tradesToKLineNew.getLowestPrice());
                        if (bigDecimal.compareTo(lowClose) < 0) {
                            tradesToKLineNew.setLowestPrice(lowCloseStr);
                        }
                    }
                }
                oldStart.plusMinutes(5);
            }
            if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
                String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_15M+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time.minusMinutes(15), DateUtils.format_ymdhm);
                currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
            }
        }
        if(m%30 == 0){
            tradesToKLineNew = new TradesToKLine();
            oldStart = time.minusMinutes(30);
            tradesToKLineNew.setVolume("0");
            tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(oldStart).getTime()));
            if(exchangeId.intValue() == 11 && "ETH/BTC".equals(transactionName)){
                LogTrace.infoKvs("5分钟K线数据", "transactionName", transactionName);
            }
            for(int i=0;i<2;i++){
                String rowKey = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_15M+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(oldStart, DateUtils.format_ymdhm);
                LogTrace.infoKvs("15分钟K线数据", "rowKey", rowKey);
                String openPriceStr;
                String highOpenStr;
                String lowCloseStr;
                String closePriceStr;
                String volumeStr;
                TradesToKLine tradesToKLine = getHbaseDataByRowKey(rowKey);
                if(tradesToKLine != null){
                    openPriceStr = tradesToKLine.getOpenPrice();
                    highOpenStr = tradesToKLine.getHighestPrice();
                    lowCloseStr = tradesToKLine.getLowestPrice();
                    closePriceStr = tradesToKLine.getClosePrice();
                    volumeStr = tradesToKLine.getVolume();
                    //交易量
                    BigDecimal volume = new BigDecimal(tradesToKLineNew.getVolume());
                    volume = volume.add(new BigDecimal(volumeStr));
                    tradesToKLineNew.setVolume(volume.toString());
                    //开盘
                    if (StringUtils.isBlank(tradesToKLineNew.getOpenPrice())) {
                        tradesToKLineNew.setOpenPrice(openPriceStr);
                    }
                    //收盘
                    tradesToKLineNew.setClosePrice(closePriceStr);
                    //最高价格
                    if (StringUtils.isBlank(tradesToKLineNew.getHighestPrice())) {
                        tradesToKLineNew.setHighestPrice(highOpenStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                        BigDecimal highOpen = new BigDecimal(tradesToKLineNew.getHighestPrice());
                        if (bigDecimal.compareTo(highOpen) > 0) {
                            tradesToKLineNew.setHighestPrice(highOpenStr);
                        }
                    }
                    //最低价格
                    if (StringUtils.isBlank(tradesToKLineNew.getLowestPrice())) {
                        tradesToKLineNew.setLowestPrice(lowCloseStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                        BigDecimal lowClose = new BigDecimal(tradesToKLineNew.getLowestPrice());
                        if (bigDecimal.compareTo(lowClose) < 0) {
                            tradesToKLineNew.setLowestPrice(lowCloseStr);
                        }
                    }
                }
                oldStart.plusMinutes(15);
            }
            if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
                String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_30M+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time.minusMinutes(30), DateUtils.format_ymdhm);
                currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
            }
        }
      /*  if(m%60 == 0){//上一个小时
            tradesToKLineNew = new TradesToKLine();
            oldStart = time.minusMinutes(60);
            tradesToKLineNew.setVolume("0");
            tradesToKLineNew.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(oldStart).getTime()));
            *//*if(exchangeId.intValue() == 11 && "ETH/BTC".equals(transactionName)){
                LogTrace.infoKvs("5分钟K线数据", "transactionName", transactionName);
            }*//*
            for(int i=0;i<2;i++){
                String rowKey = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_30M+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(oldStart, DateUtils.format_ymdhm);
                LogTrace.infoKvs("30分钟K线数据", "rowKey", rowKey);
                String openPriceStr;
                String highOpenStr;
                String lowCloseStr;
                String closePriceStr;
                String volumeStr;
                TradesToKLine tradesToKLine = getHbaseDataByRowKey(rowKey);
                if(tradesToKLine != null){
                    openPriceStr = tradesToKLine.getOpenPrice();
                    highOpenStr = tradesToKLine.getHighestPrice();
                    lowCloseStr = tradesToKLine.getLowestPrice();
                    closePriceStr = tradesToKLine.getClosePrice();
                    volumeStr = tradesToKLine.getVolume();
                    //交易量
                    BigDecimal volume = new BigDecimal(tradesToKLineNew.getVolume());
                    volume = volume.add(new BigDecimal(volumeStr));
                    tradesToKLineNew.setVolume(volume.toString());
                    //开盘
                    if (StringUtils.isBlank(tradesToKLineNew.getOpenPrice())) {
                        tradesToKLineNew.setOpenPrice(openPriceStr);
                    }
                    //收盘
                    tradesToKLineNew.setClosePrice(closePriceStr);
                    //最高价格
                    if (StringUtils.isBlank(tradesToKLineNew.getHighestPrice())) {
                        tradesToKLineNew.setHighestPrice(highOpenStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(highOpenStr);
                        BigDecimal highOpen = new BigDecimal(tradesToKLineNew.getHighestPrice());
                        if (bigDecimal.compareTo(highOpen) > 0) {
                            tradesToKLineNew.setHighestPrice(highOpenStr);
                        }
                    }
                    //最低价格
                    if (StringUtils.isBlank(tradesToKLineNew.getLowestPrice())) {
                        tradesToKLineNew.setLowestPrice(lowCloseStr);
                    } else {
                        BigDecimal bigDecimal = new BigDecimal(lowCloseStr);
                        BigDecimal lowClose = new BigDecimal(tradesToKLineNew.getLowestPrice());
                        if (bigDecimal.compareTo(lowClose) < 0) {
                            tradesToKLineNew.setLowestPrice(lowCloseStr);
                        }
                    }
                }
                oldStart =oldStart.plusMinutes(30);
            }
            if(StringUtils.isNotBlank(tradesToKLineNew.getOpenPrice())){
                String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_1H+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(time.minusMinutes(60), DateUtils.format_ymdh);
                currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLineNew);
            }
        }*/
    }
    /**
     * 15M，30M，1H时时计算k线
     * @param exchangeId
     * @param transactionName
     * @param time
     * @throws Exception
     */
    private void calculateParkMakeKAction(Integer exchangeId, String transactionName, LocalDateTime time,TradesToKLine kLine,String exchangeName) throws Exception{
        try {
            //获取当前15Mk线数据，处理
            LocalDateTime currencyH = time.minusMinutes(5);
            while (currencyH.getMinute()%15 != 0){
                currencyH = currencyH.minusMinutes(1);
            }
            //获取当前15Mk线数据，处理
            String rowKeyHead = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_15M;
            String rowKeyOld = rowKeyHead+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(currencyH, DateUtils.format_ymdhm);
            TradesToKLine tradesToKLine = getHbaseDataByRowKey(rowKeyOld);
            if(tradesToKLine != null){
                BigDecimal volume = new BigDecimal(tradesToKLine.getVolume()).add(new BigDecimal(kLine.getVolume()));
                tradesToKLine.setVolume(volume.stripTrailingZeros().toPlainString());
                tradesToKLine.setClosePrice(kLine.getClosePrice());
                //最高价格
                if (StringUtils.isBlank(tradesToKLine.getHighestPrice())) {
                    tradesToKLine.setHighestPrice(kLine.getHighestPrice());
                } else {
                    BigDecimal bigDecimal = new BigDecimal(kLine.getHighestPrice());
                    BigDecimal highOpen = new BigDecimal(tradesToKLine.getHighestPrice());
                    if (bigDecimal.compareTo(highOpen) > 0) {
                        tradesToKLine.setHighestPrice(kLine.getHighestPrice());
                    }
                }
                //最低价格
                if (StringUtils.isBlank(tradesToKLine.getLowestPrice())) {
                    tradesToKLine.setLowestPrice(kLine.getLowestPrice());
                } else {
                    BigDecimal bigDecimal = new BigDecimal(kLine.getLowestPrice());
                    BigDecimal lowClose = new BigDecimal(tradesToKLine.getLowestPrice());
                    if (bigDecimal.compareTo(lowClose) < 0) {
                        tradesToKLine.setLowestPrice(kLine.getLowestPrice());
                    }
                }
            }else{
                tradesToKLine = new TradesToKLine();
                tradesToKLine.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(currencyH).getTime()));
                tradesToKLine.setOpenPrice(kLine.getOpenPrice());
                tradesToKLine.setHighestPrice(kLine.getHighestPrice());
                tradesToKLine.setLowestPrice(kLine.getLowestPrice());
                tradesToKLine.setClosePrice(kLine.getClosePrice());
                tradesToKLine.setVolume(kLine.getVolume());
            }
            //如果不存在直接插入，否则判断当前数据与老数据的最高最低价格以及收盘价格
            if(StringUtils.isNotBlank(tradesToKLine.getOpenPrice())){
                //String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_15M+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(currencyH, DateUtils.format_ymdhm);
                currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKeyOld, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLine);
            }
            //获取当前30Mk线数据，处理
            currencyH = time.minusMinutes(5);
            while (currencyH.getMinute()%30 != 0){
                currencyH = currencyH.minusMinutes(1);
            }
            //获取当前15Mk线数据，处理
            rowKeyHead = HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_30M;
            rowKeyOld = rowKeyHead+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(currencyH, DateUtils.format_ymdhm);
            tradesToKLine = getHbaseDataByRowKey(rowKeyOld);
            if(tradesToKLine != null){
                BigDecimal volume = new BigDecimal(tradesToKLine.getVolume()).add(new BigDecimal(kLine.getVolume()));
                tradesToKLine.setVolume(volume.stripTrailingZeros().toPlainString());
                tradesToKLine.setClosePrice(kLine.getClosePrice());
                //最高价格
                if (StringUtils.isBlank(tradesToKLine.getHighestPrice())) {
                    tradesToKLine.setHighestPrice(kLine.getHighestPrice());
                } else {
                    BigDecimal bigDecimal = new BigDecimal(kLine.getHighestPrice());
                    BigDecimal highOpen = new BigDecimal(tradesToKLine.getHighestPrice());
                    if (bigDecimal.compareTo(highOpen) > 0) {
                        tradesToKLine.setHighestPrice(kLine.getHighestPrice());
                    }
                }
                //最低价格
                if (StringUtils.isBlank(tradesToKLine.getLowestPrice())) {
                    tradesToKLine.setLowestPrice(kLine.getLowestPrice());
                } else {
                    BigDecimal bigDecimal = new BigDecimal(kLine.getLowestPrice());
                    BigDecimal lowClose = new BigDecimal(tradesToKLine.getLowestPrice());
                    if (bigDecimal.compareTo(lowClose) < 0) {
                        tradesToKLine.setLowestPrice(kLine.getLowestPrice());
                    }
                }
            }else{
                tradesToKLine = new TradesToKLine();
                tradesToKLine.setTimestamp(String.valueOf(DateUtils.localDateTimeToDate(currencyH).getTime()));
                tradesToKLine.setOpenPrice(kLine.getOpenPrice());
                tradesToKLine.setHighestPrice(kLine.getHighestPrice());
                tradesToKLine.setLowestPrice(kLine.getLowestPrice());
                tradesToKLine.setClosePrice(kLine.getClosePrice());
                tradesToKLine.setVolume(kLine.getVolume());
            }
            //如果不存在直接插入，否则判断当前数据与老数据的最高最低价格以及收盘价格
            if(StringUtils.isNotBlank(tradesToKLine.getOpenPrice())){
               // String rowKey =HbaseTableUtils.K_ACTION_K_LINE_ROW_KEY_30M+"@"+exchangeId + "@" + transactionName + "@" + DateUtils.localDateTimeFormatString(currencyH, DateUtils.format_ymdhm);
                currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKeyOld, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , tradesToKLine);
            }
            //获取当前1Hk线数据，处理
            String lastDateStr = DateUtils.localDateTimeFormatString(time.minusMinutes(5), DateUtils.format_ymdh);
            String rowKey = exchangeId+"@"+transactionName+"@"+lastDateStr;
            Map<String, Object> hbaseData = hbaseBaseQueryComponent.selectData(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME
                    , rowKey);
            if(null != hbaseData && hbaseData.get("volume") != null && hbaseData.get("highOpen") != null && hbaseData.get("lowClose") != null) {
                String openPriceStr = hbaseData.get("openPrice").toString();
                String highOpenStr = hbaseData.get("highOpen").toString();
                String lowCloseStr = hbaseData.get("lowClose").toString();
                String volumeStr = hbaseData.get("volume").toString();
                BigDecimal volume = new BigDecimal(volumeStr).add(new BigDecimal(kLine.getVolume()));
                kLine.setVolume(volume.stripTrailingZeros().toPlainString());
                kLine.setOpenPrice(openPriceStr);
                //最高价格
                BigDecimal highOpenOld = new BigDecimal(highOpenStr);
                BigDecimal highOpen = new BigDecimal(kLine.getHighestPrice());
                if (highOpen.compareTo(highOpenOld) > 0) {
                    kLine.setHighestPrice(kLine.getHighestPrice());
                }
                //最低价格
                BigDecimal lowCloseOld = new BigDecimal(lowCloseStr);
                BigDecimal lowClose = new BigDecimal(kLine.getLowestPrice());
                if (lowClose.compareTo(lowCloseOld) < 0) {
                    kLine.setLowestPrice(kLine.getLowestPrice());
                }
            }
            if(kLine != null && kLine.getOpenPrice() != null){
                KLine k = new KLine();
                k.setVolume(kLine.getVolume());
                k.setOpenPrice(kLine.getOpenPrice());
                k.setHighOpen(kLine.getHighestPrice());
                k.setLowClose(kLine.getLowestPrice());
                k.setClosePrice(kLine.getClosePrice());
                k.setCreateTime(new Date());
                k.setUpdateTime(new Date());
                k.setExchangeId(exchangeId);
                k.setTransationPairName(transactionName);
                k.setExchangeName(exchangeName);
                currencyInfoHbaseComponent.insertHbaseObject(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME ,rowKey, HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_FAMILY , kLine);
            }
        }catch (Exception e){
            e.printStackTrace();
            LogTrace.error("5分钟k线后-15M，30M，1H时时计算k线error",e);
        }
    }
    /**
     * 根据row获取数据5,15等
     * @param rowKey
     * @return
     * @throws Exception
     */
    public TradesToKLine getHbaseDataByRowKey(String rowKey) throws Exception{
        /*BigDecimal maxPrice = new BigDecimal(0);
        BigDecimal minPrice = new BigDecimal(0);
        BigDecimal totalVolume = new BigDecimal(0);*/
        //String rowKey = exchange.getId()+"@"+middle.getTransactionName()+"@"+lastRowKey;
        Map<String, Object> hbaseData = hbaseBaseQueryComponent.selectData(HbaseTableUtils.CURRENCY_INFO_K_LINE_TABLE_NAME
                , rowKey);
        if(null != hbaseData && hbaseData.get("volume") != null && hbaseData.get("openPrice") != null && hbaseData.get("highestPrice") != null&& hbaseData.get("lowestPrice") != null&& hbaseData.get("closePrice") != null){
            TradesToKLine tradesToKLine = new TradesToKLine();
            /*BigDecimal volume = new BigDecimal(hbaseData.get("Volume").toString());//交易量
            BigDecimal highOpen = new BigDecimal(hbaseData.get("highestPrice").toString());//最高
            BigDecimal lowClose = new BigDecimal(hbaseData.get("lowestPrice").toString());//最低
            BigDecimal openPrice = new BigDecimal(hbaseData.get("openPrice").toString());//开盘
            BigDecimal closePrice = new BigDecimal(hbaseData.get("closePrice").toString());//最低*/
            tradesToKLine.setTimestamp(hbaseData.get("timestamp").toString());
            tradesToKLine.setVolume(hbaseData.get("volume").toString());
            tradesToKLine.setOpenPrice(hbaseData.get("openPrice").toString());
            tradesToKLine.setClosePrice(hbaseData.get("closePrice").toString());
            tradesToKLine.setHighestPrice(hbaseData.get("highestPrice").toString());
            tradesToKLine.setLowestPrice(hbaseData.get("lowestPrice").toString());
            return tradesToKLine;
            /*totalVolume = totalVolume.add(volume);

            if(maxPrice.compareTo(highOpen) < 0){
                maxPrice = highOpen;
            }
            if(minPrice.compareTo(BigDecimal.valueOf(0)) ==0){
                minPrice = lowClose;
            }else {
                if(minPrice.compareTo(lowClose) > 0){
                    minPrice = lowClose;
                }
            }*/
        }
        return null;
    }

    /**
     * 昨日资金净流入
     */
    public void calculateYesFlow() {
        try {
            //获取交易所-》交易对-》判断5/15/30/1h/4h-》分别计算维度k线
            ExchangeInfo exchangeInfo=new ExchangeInfo();
            exchangeInfo.setEnableFlag( true );
            exchangeInfo.setDeleteFlag( false );
            exchangeInfo.setKlineStatus( true );
            LocalDateTime time = LocalDateTime.now();
            List<ExchangeInfo> infoList=exchangeInfoMapper.selectByEntityList( exchangeInfo );
            if(infoList != null && infoList.size() >0){
                for(ExchangeInfo exchange : infoList){
                    ExchangeTransactionpairMiddle query = new ExchangeTransactionpairMiddle();
                    query.setExchangeId(exchange.getId());
                    //获取所有货币,组装每个需要诊断数据放到redis中
                    List<ExchangeTransactionpair> middleList = exchangeTransactionpairMiddleMapper.selectByEntityListOfKLine(query);
                    if(middleList != null && middleList.size() > 0){
                        for (ExchangeTransactionpair pair :middleList){
                            calculateMakeYesFlow(pair.getExchangeId(),pair.getTransactionName(),time,pair.getMiddleId());
                        }
                    }
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 昨日资金净流入
     * @param exchangeId
     * @param transactionName
     * @param time
     */
    private void calculateMakeYesFlow(Integer exchangeId, String transactionName, LocalDateTime time,Integer middleId) throws Exception{
        //判断mysql中是否存在昨日数据 fundsFlowMapper
        LocalDateTime yesTime = time.minusDays(1);

        //从redis获取昨日最后一条资金净流入数据，存在插入数据库中
        String redisKey24HData = RedisKeysUtils.API_SERVICE_TRADES_24H_DATA+"@"+exchangeId+"@"+transactionName;
        String redis24HData = redisServer.get(redisKey24HData);
        if(StringUtils.isBlank(redis24HData)){
            return;
        }
        List<TradesHistoryCountFive> list24Data = JSON.parseArray(redis24HData,TradesHistoryCountFive.class);
        Map<String,TradesHistoryCountFive> tradesMap = new HashMap<>();
        if(null != list24Data && list24Data.size() > 0){//转成map
            for(int i =0;i<list24Data.size();i++){
                tradesMap.put(list24Data.get(i).getDateTime(),list24Data.get(i));
            }
            int nowHour = time.getHour();
            String dateYesStr = DateUtils.localDateTimeFormatString(yesTime,DateUtils.format_ymd);
            dateYesStr +=" 23:59:59";
            Date d = DateUtils.getDate(dateYesStr,DateUtils.format_ymdhms);
            LocalDateTime yesEndTime = DateUtils.getDateTimeOfTimestamp(d.getTime());
           // time = time.minusMinutes(5);
            String timePart;
            TradesHistoryCountFive trades = null;
            String buyAmountStr = "0";
            String sellAmountStr = "0";
            String bugSellAmountStr = "0";
            String allAmountStr = "0";
            FundsFlow funds = new FundsFlow();
            for(int i=0;i<288;i++) {//循环288次，共24小时,5分钟一条
                int minu = yesEndTime.getMinute() / 5;
                timePart = DateUtils.localDateTimeFormatString(yesEndTime, DateUtils.format_ymdh);
                timePart += minu < 10 ? "0" + minu : minu;
                trades = tradesMap.get(timePart);
                if(trades!=null && trades.getAmountNew() != null){
                    buyAmountStr = trades.getBuyAmountNew().setScale(2,BigDecimal.ROUND_DOWN).stripTrailingZeros().toPlainString();
                    sellAmountStr = trades.getSellAmountNew().setScale(2,BigDecimal.ROUND_DOWN).stripTrailingZeros().toPlainString();
                    bugSellAmountStr = trades.getBuyAmountNew().subtract(trades.getSellAmountNew()).setScale(2,BigDecimal.ROUND_DOWN).stripTrailingZeros().toPlainString();
                    allAmountStr = trades.getBuyAmountNew().add(trades.getSellAmountNew()).setScale(2,BigDecimal.ROUND_DOWN).stripTrailingZeros().toPlainString();
                    funds.setFundsInflow(buyAmountStr);
                    funds.setFundsOutflow(sellAmountStr);
                    funds.setFundsOnlyflow(bugSellAmountStr);
                    funds.setFundsTotal(allAmountStr);
                    break;
                }
                yesEndTime = yesEndTime.minusMinutes(5);
            }
            if(StringUtils.isNotBlank(funds.getFundsTotal())){//插入
                FundsFlow fundsFlow = new FundsFlow();
                fundsFlow.setExchangeId(exchangeId);
                fundsFlow.setTransactionName(transactionName);
                fundsFlow.setExchangeTranMiddleId(middleId);
                fundsFlow.setFundsDate(DateUtils.localDateTimeFormatString(yesTime,DateUtils.format_ymd));
                fundsFlow = fundsFlowMapper.selectByEntity(fundsFlow);
                //已经存在更新，否则插入
                if(fundsFlow != null){//更新
                    fundsFlow.setFundsInflow(funds.getFundsInflow());
                    fundsFlow.setFundsOutflow(funds.getFundsOutflow());
                    fundsFlow.setFundsOnlyflow(funds.getFundsOnlyflow());
                    fundsFlow.setFundsTotal(funds.getFundsTotal());
                    fundsFlow.setCreateTime(new Date());
                    fundsFlowMapper.updateByPrimaryKey(fundsFlow);
                }else{//插入
                    funds.setExchangeId(exchangeId);
                    funds.setTransactionName(transactionName);
                    funds.setExchangeTranMiddleId(middleId);
                    funds.setFundsDate(DateUtils.localDateTimeFormatString(yesTime,DateUtils.format_ymd));
                    funds.setCreateTime(new Date());
                    fundsFlowMapper.insert(funds);
                }
            }
        }
    }

    public void initExchangePingTaiB() {
        //查询所有交易所，循环调取akka
        ExchangeInfo exchangeInfo=new ExchangeInfo();
        exchangeInfo.setEnableFlag( true );
        exchangeInfo.setDeleteFlag( false );
        exchangeInfo.setKlineStatus( true );
        List<ExchangeInfo> infoList=exchangeInfoMapper.selectByEntityList( exchangeInfo );
        Integer start = 0;
        Integer end = 0;
        for(ExchangeInfo exchangeInfo1 : infoList){
            if("Hadax".equals(exchangeInfo1.getExchangeName())){
                start = exchangeInfo1.getId();
            }
            if("huobipro".equals(exchangeInfo1.getExchangeName())){
                end = exchangeInfo1.getId();
            }
        }
        if(start.intValue() != 0 && end.intValue() != 0){
            redisServer.set(RedisKeyApp.EXCHANGE_PING_TAI_B_CONVERT+start,end);
        }
    }

    public void initExchangePingTaiB(Integer start, Integer end) {
        if(start.intValue() != 0 && end.intValue() != 0){
            redisServer.set(RedisKeyApp.EXCHANGE_PING_TAI_B_CONVERT+start,end);
        }
    }
}
